package com.bigfans.cartservice.listener;

import com.bigfans.cartservice.api.clients.CatalogServiceClient;
import com.bigfans.cartservice.model.Product;
import com.bigfans.cartservice.model.ProductSpec;
import com.bigfans.cartservice.service.ProductService;
import com.bigfans.cartservice.service.ProductSpecService;
import com.bigfans.framework.kafka.KafkaConsumerBean;
import com.bigfans.framework.kafka.KafkaListener;
import com.bigfans.model.event.ProductCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * @author lichong
 * @create 2018-02-03 下午6:11
 **/
@KafkaConsumerBean
@Component
public class ProductListener {

    private Logger logger = LoggerFactory.getLogger(ProductListener.class);

    @Autowired
    private ProductService productService;
    @Autowired
    private ProductSpecService productSpecService;
    @Autowired
    private CatalogServiceClient catalogServiceClient;

    @KafkaListener
    public void on(ProductCreatedEvent event){
        try{
            String prodId = event.getId();
            CompletableFuture<Product> prodFuture = catalogServiceClient.getProductById(prodId);
            CompletableFuture<List<ProductSpec>> specsFuture = catalogServiceClient.getSpecsByProductId(prodId);

            CompletableFuture.allOf(prodFuture , specsFuture).join();

            Product product = prodFuture.get();
            productService.create(product);

            List<ProductSpec> specs = specsFuture.get();
            productSpecService.batchCreate(specs);
        }catch (Exception e){
            logger.error("");
        }
    }

}
